package com.oct.ywsc;

import com.alibaba.fastjson2.JSONObject;
import com.oct.ywsc.core.JsonUtil;
import com.oct.ywsc.job.CustomerKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class Main {

    public static void main(String[] args) {
        Properties props = new Properties();
        //1.指定Kafaka集群的ip地址和端口号
        props.put("bootstrap.servers", "10.0.11.47:9092");
        //2.等待所有副本节点的应答
        props.put("acks", "all");
        //3.消息发送最大尝试次数
        props.put("retries", 1);
        //4.指定一批消息处理次数
        props.put("batch.size", 16384);
        //5.指定请求延时
        props.put("linger.ms", 1);
        //6.指定缓存区内存大小
        props.put("buffer.memory", 33554432);
        //7.设置key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //8.设置value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 9、生产数据
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);


        String msg = "{\"before\":{\"BMSAH\":\"XXXXXXXXX\",\"id\":\"2\"},\"op_type\":\"D\",\"op_ts\":1742458833000,\"table\":\"t_student\",\"scn\":7273468186}";

        int i =0;
        for (; i < 100; i++){
            producer.send(new ProducerRecord<String, String>("ywsc", i+"", msg));

//            try {
//                Thread.currentThread().sleep(1000);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }

        }

        producer.close();
    }
}
